package architect;

import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.io.jdbc.JDBCInputFormat;
import org.apache.flink.api.java.typeutils.RowTypeInfo;
import org.apache.flink.types.Row;
import org.apache.parquet.schema.PrimitiveType;

import java.util.LinkedHashMap;

import static org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName.BINARY;

/**
 * @title: FlinkMySQL
 * @description:
 * @author: guan.xin
 * @create: 2022-04-07 15:40
 * @since: JDK8
 */
public class FlinkMySQL {

    public static void main(String[] args) throws Exception {
        // JobExecutionException
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

        DataSet<Row> ds = env.createInput(JDBCInputFormat.buildJDBCInputFormat()
                // 配置数据库连接信息
                .setDrivername("com.mysql.cj.jdbc.Driver")
                // 数据库URL
                .setDBUrl("jdbc:mysql://localhost:3306/fcsdb?serverTimezone=GMT%2B8&useSSL=false")
                // 用户名
                .setUsername("root")
                // 登录密码
                .setPassword("123456")
                // 需要执行的SQL语句
                .setQuery("SELECT user_id, user_name, real_name, nick_name, user_type FROM fcs_user_info")
                // 设置查询的列的类型
                .setRowTypeInfo(new RowTypeInfo(
                        BasicTypeInfo.STRING_TYPE_INFO,
                        BasicTypeInfo.STRING_TYPE_INFO,
                        BasicTypeInfo.STRING_TYPE_INFO,
                        BasicTypeInfo.STRING_TYPE_INFO,
                        BasicTypeInfo.STRING_TYPE_INFO))
                .finish());

//        DataSet<Row> ds = env.createInput(JDBCInputFormat.buildJDBCInputFormat()
//                // 配置数据库连接信息
//                .setDrivername("com.mysql.cj.jdbc.Driver")
//                // 数据库URL
//                .setDBUrl("jdbc:mysql://w-kh-sit-oms4-mysql8-public.service.testdb:3306/db_oms4_som01")
//                // 用户名
//                .setUsername("u_oms4_som")
//                // 登录密码
//                .setPassword("Oms4321com")
//                // 需要执行的SQL语句
//                .setQuery("SELECT COLUMN_NAME, DATA_TYPE AS TYPE FROM information_schema.`COLUMNS` WHERE TABLE_SCHEMA = 'db_oms4_som01' and TABLE_NAME = 'so_do_order'")
//                // 设置查询的列的类型
//                .setRowTypeInfo(new RowTypeInfo(
//                        BasicTypeInfo.STRING_TYPE_INFO,
//                        BasicTypeInfo.STRING_TYPE_INFO
//                ))
//                .finish());

        // 转换为Student类型
//        DataSet<UserInfo> dataMap = dataInput.map((MapFunction<Row, UserInfo>) row -> new UserInfo(
//                (String) row.getField(0),
//                (String) row.getField(1),
//                (String) row.getField(2),
//                (String) row.getField(3),
//                (String) row.getField(4)
//        ));

        LinkedHashMap<String, PrimitiveType.PrimitiveTypeName> map = new LinkedHashMap<>();
        map.put("COLUMN_NAME", BINARY);
        map.put("DATA_TYPE", BINARY);
        map.put("real_name", BINARY);
        map.put("nick_name", BINARY);
        map.put("user_type", BINARY);


        // ToDo 3.sink
        //ds.print();
        ds.output(new ParquetOutputFormat("/Users/guan.xin/Gitee/Python/pythonProject/parquet_files/allin/so_do_order.parquet", "so_do_order", map));

//        Types.MessageTypeBuilder messageTypeBuilder = Types.buildMessage();
//        map.forEach((k, v) -> {
//            messageTypeBuilder.required(v).as(UTF8).named(k);
//        });
//        MessageType messageType = messageTypeBuilder.named("user_info");
//
//        SimpleGroupFactory factory = new SimpleGroupFactory(messageType);
//
//        try {
//            ParquetWriter<Group> writer = ExampleParquetWriter.builder(new Path("/Users/guan.xin/Gitee/Python/pythonProject/parquet_files/allin/user_info.parquet"))
//                    .withWriteMode(ParquetFileWriter.Mode.OVERWRITE)
//                    .withType(messageType)
//                    .build();
//
//            ds.collect().forEach(row -> {
//                Group group = factory.newGroup();
//                group.add(0, row.getFieldAs(0).toString());
//                group.add(1, row.getFieldAs(1).toString());
//                group.add(2, row.getFieldAs(2).toString());
//                group.add(3, row.getFieldAs(3).toString());
//                group.add(4, row.getFieldAs(4).toString());
//
//                try {
//                    writer.write(group);
//                } catch (IOException e) {
//                    e.printStackTrace();
//                }
//            });
//
//            writer.close();
//        } catch (IOException e) {
//            e.printStackTrace();
//        }

        // ToDo 4.execute
        env.execute();
    }

    public static class UserInfo {
        private String userId;
        private String userName;
        private String realName;
        private String nickName;
        private String userType;

        public UserInfo(String userId, String userName, String realName, String nickName, String userType) {
            this.userId = userId;
            this.userName = userName;
            this.realName = realName;
            this.nickName = nickName;
            this.userType = userType;
        }

        public String getUserId() {
            return userId;
        }

        public UserInfo setUserId(String userId) {
            this.userId = userId;
            return this;
        }

        public String getUserName() {
            return userName;
        }

        public UserInfo setUserName(String userName) {
            this.userName = userName;
            return this;
        }

        public String getRealName() {
            return realName;
        }

        public UserInfo setRealName(String realName) {
            this.realName = realName;
            return this;
        }

        public String getNickName() {
            return nickName;
        }

        public UserInfo setNickName(String nickName) {
            this.nickName = nickName;
            return this;
        }

        public String getUserType() {
            return userType;
        }

        public UserInfo setUserType(String userType) {
            this.userType = userType;
            return this;
        }

        @Override
        public String toString() {
            return "UserInfo{" +
                    "userId='" + userId + '\'' +
                    ", userName='" + userName + '\'' +
                    ", realName='" + realName + '\'' +
                    ", nickName='" + nickName + '\'' +
                    ", userType='" + userType + '\'' +
                    '}';
        }
    }
}
